package com.lsx143.wordcount.day3.sink;

import com.lsx143.wordcount.bean.WaterSensor;
import com.lsx143.wordcount.day2.RandomWaterSensor;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;

public class KafkaSink {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);
        DataStreamSource<WaterSensor> dss = env.addSource(new RandomWaterSensor(RandomWaterSensor.RUN_MODE.RANDOM));
        dss
                .map(WaterSensor::toString)
                .addSink(new FlinkKafkaProducer<>(
                        "hadoop102:9092",
                        "waterSenor",
                        new SimpleStringSchema()
                ));
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
